/*-------------------------------------------------------------------------
 *
 * be-fsstubs.c
 *	  用于大对象的打开/关闭/读取/写入操作的内置函数
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/libpq/be-fsstubs.c
 *
 * NOTES
 *	  这应该移动到更合适的地方。它在这里是因为没有更好的地方。
 *
 *	  这些函数将LargeObjectDesc结构存储在一个私有MemoryContext中，
 *	  这意味着大对象描述符会在我们在事务结束时销毁上下文之前保留。
 *	  可以延长上下文的生存期，以使LO FDs在事务之间保持有效
 *	  （例如，我们可以在看到没有FD保持打开时仅释放上下文）。
 *	  但我们需要额外的状态，以便在事务中止时采取正确的行动。
 *	  在中止的事务期间打开的FD仍需要关闭，因为它们可能根本不指向
 *	  有效的关系。如果LO在事务之间保持打开，锁定语义也是一个有趣
 *	  的问题。目前，我们将坚持LO FD的现有文档语义：它们仅在事务内
 *	  有效。
 *
 *	  自PostgreSQL 8.0起，上述担忧已不再相关，实际上非常容易
 *	  允许LO FDs在事务之间保持打开。 （快照相关性仍然是一个问题。）
 *	  然而，向后兼容性建议我们应坚持现状。
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include <fcntl.h>
#include <sys/stat.h>
#include <unistd.h>

#include "access/xact.h"
#include "libpq/be-fsstubs.h"
#include "libpq/libpq-fs.h"
#include "miscadmin.h"
#include "storage/fd.h"
#include "storage/large_object.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"

/* 定义此项以启用调试日志 */
/* #define FSDB 1 */
/* lo_import/lo_export 传输的块大小 */
#define BUFSIZE			8192

/*
 * LO "FD" 是 cookies 数组的索引。
 *
 * 非空条目是指向在 LO 私有内存上下文 "fscxt" 中分配的 LargeObjectDesc 的指针。
 * cookies 数组本身也是在该上下文中动态分配的。其当前分配大小为 cookies_size 条目，
 * 其中任何未使用的条目将为 NULL。
 */
static LargeObjectDesc **cookies = NULL;
static int	cookies_size = 0;

static bool lo_cleanup_needed = false;
static MemoryContext fscxt = NULL;

static int	fc_newLOfd(void);
static void fc_closeLOfd(int fc_fd);
static Oid	fc_lo_import_internal(text *fc_filename, Oid fc_lobjOid);


/*****************************************************************************
 *	大型对象的文件接口
 *****************************************************************************/

Datum be_lo_open(PG_FUNCTION_ARGS)
{
	Oid			fc_lobjId = PG_GETARG_OID(0);
	int32		fc_mode = PG_GETARG_INT32(1);
	LargeObjectDesc *fc_lobjDesc;
	int			fc_fd;

#ifdef FSDB
	elog(DEBUG4, "lo_open(%u,%d)", fc_lobjId, fc_mode);
#endif

	/*
	 * 首先分配一个大型对象描述符。如果这是该事务中打开的第一个 LO，
	 * 这也会创建 'fscxt'。
	 */
	fc_fd = fc_newLOfd();

	fc_lobjDesc = inv_open(fc_lobjId, fc_mode, fscxt);
	fc_lobjDesc->subid = GetCurrentSubTransactionId();

	/*
	 * 我们必须将快照注册到 TopTransaction 的 resowner 中，以便在 LO 关闭之前保持其存在，
	 * 而不是在当前门户关闭之前。
	 */
	if (fc_lobjDesc->snapshot)
		fc_lobjDesc->snapshot = RegisterSnapshotOnOwner(fc_lobjDesc->snapshot,
													 TopTransactionResourceOwner);

	Assert(cookies[fc_fd] == NULL);
	cookies[fc_fd] = fc_lobjDesc;

	PG_RETURN_INT32(fc_fd);
}

Datum be_lo_close(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);

	if (fc_fd < 0 || fc_fd >= cookies_size || cookies[fc_fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fc_fd)));

#ifdef FSDB
	elog(DEBUG4, "lo_close(%d)", fc_fd);
#endif

	fc_closeLOfd(fc_fd);

	PG_RETURN_INT32(0);
}


/*****************************************************************************
 *	裸读/写操作——这些不能通过 fmgr 调用！
 *
 *	我们假设大型对象支持面向字节的读取和寻址，以便我们的工作更加轻松。
 *
 *****************************************************************************/

int lo_read(int fc_fd, char *fc_buf, int fc_len)
{
	int			fc_status;
	LargeObjectDesc *fc_lobj;

	if (fc_fd < 0 || fc_fd >= cookies_size || cookies[fc_fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fc_fd)));
	fc_lobj = cookies[fc_fd];

	/*
	 * 检查状态。inv_read() 无论如何都会抛出错误，但我们想要的错误是关于 FD 状态的，而不是底层权限；
	 * 权限可能存在，但用户忘记请求读取模式。
	 */
	if ((fc_lobj->flags & IFS_RDLOCK) == 0)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("large object descriptor %d was not opened for reading",
						fc_fd)));

	fc_status = inv_read(fc_lobj, fc_buf, fc_len);

	return fc_status;
}

int lo_write(int fc_fd, const char *fc_buf, int fc_len)
{
	int			fc_status;
	LargeObjectDesc *fc_lobj;

	if (fc_fd < 0 || fc_fd >= cookies_size || cookies[fc_fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fc_fd)));
	fc_lobj = cookies[fc_fd];

	/* 请参见 lo_read() 中的注释 */
	if ((fc_lobj->flags & IFS_WRLOCK) == 0)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("large object descriptor %d was not opened for writing",
						fc_fd)));

	fc_status = inv_write(fc_lobj, fc_buf, fc_len);

	return fc_status;
}

Datum be_lo_lseek(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);
	int32		fc_offset = PG_GETARG_INT32(1);
	int32		fc_whence = PG_GETARG_INT32(2);
	int64		fc_status;

	if (fc_fd < 0 || fc_fd >= cookies_size || cookies[fc_fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fc_fd)));

	fc_status = inv_seek(cookies[fc_fd], fc_offset, fc_whence);

	/* 防止结果溢出 */
	if (fc_status != (int32) fc_status)
		ereport(ERROR,
				(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
				 errmsg("lo_lseek result out of range for large-object descriptor %d",
						fc_fd)));

	PG_RETURN_INT32((int32) fc_status);
}

Datum be_lo_lseek64(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);
	int64		fc_offset = PG_GETARG_INT64(1);
	int32		fc_whence = PG_GETARG_INT32(2);
	int64		fc_status;

	if (fc_fd < 0 || fc_fd >= cookies_size || cookies[fc_fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fc_fd)));

	fc_status = inv_seek(cookies[fc_fd], fc_offset, fc_whence);

	PG_RETURN_INT64(fc_status);
}

Datum be_lo_creat(PG_FUNCTION_ARGS)
{
	Oid			fc_lobjId;

	lo_cleanup_needed = true;
	fc_lobjId = inv_create(InvalidOid);

	PG_RETURN_OID(fc_lobjId);
}

Datum be_lo_create(PG_FUNCTION_ARGS)
{
	Oid			fc_lobjId = PG_GETARG_OID(0);

	lo_cleanup_needed = true;
	fc_lobjId = inv_create(fc_lobjId);

	PG_RETURN_OID(fc_lobjId);
}

Datum be_lo_tell(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);
	int64		fc_offset;

	if (fc_fd < 0 || fc_fd >= cookies_size || cookies[fc_fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fc_fd)));

	fc_offset = inv_tell(cookies[fc_fd]);

	/* 防止结果溢出 */
	if (fc_offset != (int32) fc_offset)
		ereport(ERROR,
				(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
				 errmsg("lo_tell result out of range for large-object descriptor %d",
						fc_fd)));

	PG_RETURN_INT32((int32) fc_offset);
}

Datum be_lo_tell64(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);
	int64		fc_offset;

	if (fc_fd < 0 || fc_fd >= cookies_size || cookies[fc_fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fc_fd)));

	fc_offset = inv_tell(cookies[fc_fd]);

	PG_RETURN_INT64(fc_offset);
}

Datum be_lo_unlink(PG_FUNCTION_ARGS)
{
	Oid			fc_lobjId = PG_GETARG_OID(0);

	/*
	 * 必须是大型对象的所有者。在 inv_drop() 中检查这一点会更清晰，
	 * 但我们希望在关闭相关 FD 之前抛出错误，而不是之后。
	 */
	if (!lo_compat_privileges &&
		!pg_largeobject_ownercheck(fc_lobjId, GetUserId()))
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("must be owner of large object %u", fc_lobjId)));

	/*
	 * 如果有任何开放的 LO FD 引用该 ID，则关闭它们。
	 */
	if (fscxt != NULL)
	{
		int			fc_i;

		for (fc_i = 0; fc_i < cookies_size; fc_i++)
		{
			if (cookies[fc_i] != NULL && cookies[fc_i]->id == fc_lobjId)
				fc_closeLOfd(fc_i);
		}
	}

	/*
	 * inv_drop 并不需要结束事务的清理，因此我们不需要设置 lo_cleanup_needed。
	 */
	PG_RETURN_INT32(inv_drop(fc_lobjId));
}

/*****************************************************************************
 *	使用 bytea 进行读/写
 *****************************************************************************/

Datum be_loread(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);
	int32		fc_len = PG_GETARG_INT32(1);
	bytea	   *fc_retval;
	int			fc_totalread;

	if (fc_len < 0)
		fc_len = 0;

	fc_retval = (bytea *) palloc(VARHDRSZ + fc_len);
	fc_totalread = lo_read(fc_fd, VARDATA(fc_retval), fc_len);
	SET_VARSIZE(fc_retval, fc_totalread + VARHDRSZ);

	PG_RETURN_BYTEA_P(fc_retval);
}

Datum be_lowrite(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);
	bytea	   *fc_wbuf = PG_GETARG_BYTEA_PP(1);
	int			fc_bytestowrite;
	int			fc_totalwritten;

	fc_bytestowrite = VARSIZE_ANY_EXHDR(fc_wbuf);
	fc_totalwritten = lo_write(fc_fd, VARDATA_ANY(fc_wbuf), fc_bytestowrite);
	PG_RETURN_INT32(fc_totalwritten);
}

/*****************************************************************************
 *	大型对象的导入/导出
 *****************************************************************************/

/*
 * lo_import -
 *	  将文件导入为（反转）大型对象。
 */
Datum be_lo_import(PG_FUNCTION_ARGS)
{
	text	   *fc_filename = PG_GETARG_TEXT_PP(0);

	PG_RETURN_OID(fc_lo_import_internal(fc_filename, InvalidOid));
}

/*
 * lo_import_with_oid -
 *	  将文件导入为（反转）大型对象并指定 oid。
 */
Datum be_lo_import_with_oid(PG_FUNCTION_ARGS)
{
	text	   *fc_filename = PG_GETARG_TEXT_PP(0);
	Oid			fc_oid = PG_GETARG_OID(1);

	PG_RETURN_OID(fc_lo_import_internal(fc_filename, fc_oid));
}

static Oid fc_lo_import_internal(text *fc_filename, Oid fc_lobjOid)
{
	int			fc_fd;
	int			fc_nbytes,
				fc_tmp PG_USED_FOR_ASSERTS_ONLY;
	char		fc_buf[BUFSIZE];
	char		fc_fnamebuf[MAXPGPATH];
	LargeObjectDesc *fc_lobj;
	Oid			fc_oid;

	/*
	 * 打开要读取的文件
	 */
	text_to_cstring_buffer(fc_filename, fc_fnamebuf, sizeof(fc_fnamebuf));
	fc_fd = OpenTransientFile(fc_fnamebuf, O_RDONLY | PG_BINARY);
	if (fc_fd < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not open server file \"%s\": %m",
						fc_fnamebuf)));

	/*
	 * 创建一个反转对象
	 */
	lo_cleanup_needed = true;
	fc_oid = inv_create(fc_lobjOid);

	/*
	 * 从文件系统读取并写入反转对象
	 */
	fc_lobj = inv_open(fc_oid, INV_WRITE, CurrentMemoryContext);

	while ((fc_nbytes = read(fc_fd, fc_buf, BUFSIZE)) > 0)
	{
		fc_tmp = inv_write(fc_lobj, fc_buf, fc_nbytes);
		Assert(fc_tmp == fc_nbytes);
	}

	if (fc_nbytes < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not read server file \"%s\": %m",
						fc_fnamebuf)));

	inv_close(fc_lobj);

	if (CloseTransientFile(fc_fd) != 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not close file \"%s\": %m",
						fc_fnamebuf)));

	return fc_oid;
}

/*
 * lo_export -
 *	  导出（反转）大型对象。
 */
Datum be_lo_export(PG_FUNCTION_ARGS)
{
	Oid			fc_lobjId = PG_GETARG_OID(0);
	text	   *fc_filename = PG_GETARG_TEXT_PP(1);
	int			fc_fd;
	int			fc_nbytes,
				fc_tmp;
	char		fc_buf[BUFSIZE];
	char		fc_fnamebuf[MAXPGPATH];
	LargeObjectDesc *fc_lobj;
	mode_t		fc_oumask;

	/*
	 * 打开反转对象（无需测试失败）
	 */
	lo_cleanup_needed = true;
	fc_lobj = inv_open(fc_lobjId, INV_READ, CurrentMemoryContext);

	/*
	 * 打开要写入的文件
	 *
	 * 注意：我们将后端的正常 077 umask 降低到稍微友好的
	 * 022。此代码曾经将其降低到 0，但创建
	 * 世界可写的导出文件似乎不明智。
	 */
	text_to_cstring_buffer(fc_filename, fc_fnamebuf, sizeof(fc_fnamebuf));
	fc_oumask = umask(S_IWGRP | S_IWOTH);
	PG_TRY();
	{
		fc_fd = OpenTransientFilePerm(fc_fnamebuf, O_CREAT | O_WRONLY | O_TRUNC | PG_BINARY,
								   S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
	}
	PG_FINALLY();
	{
		umask(fc_oumask);
	}
	PG_END_TRY();
	if (fc_fd < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not create server file \"%s\": %m",
						fc_fnamebuf)));

	/*
	 * 从反转文件中读取并写入文件系统
	 */
	while ((fc_nbytes = inv_read(fc_lobj, fc_buf, BUFSIZE)) > 0)
	{
		fc_tmp = write(fc_fd, fc_buf, fc_nbytes);
		if (fc_tmp != fc_nbytes)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not write server file \"%s\": %m",
							fc_fnamebuf)));
	}

	if (CloseTransientFile(fc_fd) != 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not close file \"%s\": %m",
						fc_fnamebuf)));

	inv_close(fc_lobj);

	PG_RETURN_INT32(1);
}

/*
 * lo_truncate -
 *	  将大对象截断到指定长度
 */
static void fc_lo_truncate_internal(int32 fc_fd, int64 fc_len)
{
	LargeObjectDesc *fc_lobj;

	if (fc_fd < 0 || fc_fd >= cookies_size || cookies[fc_fd] == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("invalid large-object descriptor: %d", fc_fd)));
	fc_lobj = cookies[fc_fd];

	/* 请参见 lo_read() 中的注释 */
	if ((fc_lobj->flags & IFS_WRLOCK) == 0)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("large object descriptor %d was not opened for writing",
						fc_fd)));

	inv_truncate(fc_lobj, fc_len);
}

Datum be_lo_truncate(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);
	int32		fc_len = PG_GETARG_INT32(1);

	fc_lo_truncate_internal(fc_fd, fc_len);
	PG_RETURN_INT32(0);
}

Datum be_lo_truncate64(PG_FUNCTION_ARGS)
{
	int32		fc_fd = PG_GETARG_INT32(0);
	int64		fc_len = PG_GETARG_INT64(1);

	fc_lo_truncate_internal(fc_fd, fc_len);
	PG_RETURN_INT32(0);
}

/*
 * AtEOXact_LargeObject -
 *		 为事务提交准备大对象
 */
void AtEOXact_LargeObject(bool fc_isCommit)
{
	int			fc_i;

	if (!lo_cleanup_needed)
		return;					/* 此xact中没有LO操作 */

	/*
	 * 关闭LO文件描述符并清空cookie数组，使得LO文件描述符不再有效。
	 * 内存上下文和持有它们的资源所有者在
	 * 事务结束时会消失，但在提交时，我们需要关闭它们以
	 * 避免关于提交时泄漏资源的警告。在中止时，我们可以跳过
	 * 此步骤。
	 */
	if (fc_isCommit)
	{
		for (fc_i = 0; fc_i < cookies_size; fc_i++)
		{
			if (cookies[fc_i] != NULL)
				fc_closeLOfd(fc_i);
		}
	}

	/* 实际上不需要pfree，因为我们即将销毁上下文 */
	cookies = NULL;
	cookies_size = 0;

	/* 释放LO内存上下文以防止永久内存泄漏。 */
	if (fscxt)
		MemoryContextDelete(fscxt);
	fscxt = NULL;

	/* 也给 inv_api.c 一个清理的机会 */
	close_lo_relation(fc_isCommit);

	lo_cleanup_needed = false;
}

/*
 * AtEOSubXact_LargeObject
 *		在子事务提交/中止时处理大对象
 *
 * 在提交的子事务中创建/打开的LO重新分配
 * 给父子事务。在中止时，直接关闭它们。
 */
void AtEOSubXact_LargeObject(bool fc_isCommit, SubTransactionId fc_mySubid,
						SubTransactionId fc_parentSubid)
{
	int			fc_i;

	if (fscxt == NULL)			/* 此xact中没有LO操作 */
		return;

	for (fc_i = 0; fc_i < cookies_size; fc_i++)
	{
		LargeObjectDesc *fc_lo = cookies[fc_i];

		if (fc_lo != NULL && fc_lo->subid == fc_mySubid)
		{
			if (fc_isCommit)
				fc_lo->subid = fc_parentSubid;
			else
				fc_closeLOfd(fc_i);
		}
	}
}

/*****************************************************************************
 *	此文件的支持例程
 *****************************************************************************/

static int fc_newLOfd(void)
{
	int			fc_i,
				fc_newsize;

	lo_cleanup_needed = true;
	if (fscxt == NULL)
		fscxt = AllocSetContextCreate(TopMemoryContext,
									  "Filesystem",
									  ALLOCSET_DEFAULT_SIZES);

	/* 尝试找到一个空闲的槽 */
	for (fc_i = 0; fc_i < cookies_size; fc_i++)
	{
		if (cookies[fc_i] == NULL)
			return fc_i;
	}

	/* 没有空闲槽，因此增大数组 */
	if (cookies_size <= 0)
	{
		/* 第一次通过，随意创建一个64元素的数组 */
		fc_i = 0;
		fc_newsize = 64;
		cookies = (LargeObjectDesc **)
			MemoryContextAllocZero(fscxt, fc_newsize * sizeof(LargeObjectDesc *));
		cookies_size = fc_newsize;
	}
	else
	{
		/* 将数组大小翻倍 */
		fc_i = cookies_size;
		fc_newsize = cookies_size * 2;
		cookies = (LargeObjectDesc **)
			repalloc(cookies, fc_newsize * sizeof(LargeObjectDesc *));
		MemSet(cookies + cookies_size, 0,
			   (fc_newsize - cookies_size) * sizeof(LargeObjectDesc *));
		cookies_size = fc_newsize;
	}

	return fc_i;
}

static void fc_closeLOfd(int fc_fd)
{
	LargeObjectDesc *fc_lobj;

	/*
	 * 确保我们不会在某些原因导致错误时尝试双重释放。
	 * 泄漏比崩溃更好。
	 */
	fc_lobj = cookies[fc_fd];
	cookies[fc_fd] = NULL;

	if (fc_lobj->snapshot)
		UnregisterSnapshotFromOwner(fc_lobj->snapshot,
									TopTransactionResourceOwner);
	inv_close(fc_lobj);
}

/*****************************************************************************
 *	面向SQL调用者的包装
 *****************************************************************************/

/*
 * 在LO中读取 [offset, offset+nbytes)；当 nbytes 为 -1 时，读取到结尾。
 */
static bytea * fc_lo_get_fragment_internal(Oid fc_loOid, int64 fc_offset, int32 fc_nbytes)
{
	LargeObjectDesc *fc_loDesc;
	int64		fc_loSize;
	int64		fc_result_length;
	int			fc_total_read PG_USED_FOR_ASSERTS_ONLY;
	bytea	   *fc_result = NULL;

	lo_cleanup_needed = true;
	fc_loDesc = inv_open(fc_loOid, INV_READ, CurrentMemoryContext);

	/*
	 * 计算我们实际将读取的字节数，适应 nbytes == -1
	 * 和超出LO结尾的读取。
	 */
	fc_loSize = inv_seek(fc_loDesc, 0, SEEK_END);
	if (fc_loSize > fc_offset)
	{
		if (fc_nbytes >= 0 && fc_nbytes <= fc_loSize - fc_offset)
			fc_result_length = fc_nbytes; /* 请求完全位于LO内部 */
		else
			fc_result_length = fc_loSize - fc_offset;	/* 调整到LO的末尾 */
	}
	else
		fc_result_length = 0;		/* 请求完全位于LO外部 */

	/*
	 * 从 loSize 计算的 result_length 可能无法适应 size_t。
	 * 检查该大小是否符合此后强制的大小限制。
	 */
	if (fc_result_length > MaxAllocSize - VARHDRSZ)
		ereport(ERROR,
				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
				 errmsg("large object read request is too large")));

	fc_result = (bytea *) palloc(VARHDRSZ + fc_result_length);

	inv_seek(fc_loDesc, fc_offset, SEEK_SET);
	fc_total_read = inv_read(fc_loDesc, VARDATA(fc_result), fc_result_length);
	Assert(fc_total_read == fc_result_length);
	SET_VARSIZE(fc_result, fc_result_length + VARHDRSZ);

	inv_close(fc_loDesc);

	return fc_result;
}

/*
 * 读取整个LO
 */
Datum be_lo_get(PG_FUNCTION_ARGS)
{
	Oid			fc_loOid = PG_GETARG_OID(0);
	bytea	   *fc_result;

	fc_result = fc_lo_get_fragment_internal(fc_loOid, 0, -1);

	PG_RETURN_BYTEA_P(fc_result);
}

/*
 * 读取LO中的范围
 */
Datum be_lo_get_fragment(PG_FUNCTION_ARGS)
{
	Oid			fc_loOid = PG_GETARG_OID(0);
	int64		fc_offset = PG_GETARG_INT64(1);
	int32		fc_nbytes = PG_GETARG_INT32(2);
	bytea	   *fc_result;

	if (fc_nbytes < 0)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("requested length cannot be negative")));

	fc_result = fc_lo_get_fragment_internal(fc_loOid, fc_offset, fc_nbytes);

	PG_RETURN_BYTEA_P(fc_result);
}

/*
 * 根据bytea参数创建LO，并给出初始内容
 */
Datum be_lo_from_bytea(PG_FUNCTION_ARGS)
{
	Oid			fc_loOid = PG_GETARG_OID(0);
	bytea	   *fc_str = PG_GETARG_BYTEA_PP(1);
	LargeObjectDesc *fc_loDesc;
	int			fc_written PG_USED_FOR_ASSERTS_ONLY;

	lo_cleanup_needed = true;
	fc_loOid = inv_create(fc_loOid);
	fc_loDesc = inv_open(fc_loOid, INV_WRITE, CurrentMemoryContext);
	fc_written = inv_write(fc_loDesc, VARDATA_ANY(fc_str), VARSIZE_ANY_EXHDR(fc_str));
	Assert(fc_written == VARSIZE_ANY_EXHDR(fc_str));
	inv_close(fc_loDesc);

	PG_RETURN_OID(fc_loOid);
}

/*
 * 更新LO中的范围
 */
Datum be_lo_put(PG_FUNCTION_ARGS)
{
	Oid			fc_loOid = PG_GETARG_OID(0);
	int64		fc_offset = PG_GETARG_INT64(1);
	bytea	   *fc_str = PG_GETARG_BYTEA_PP(2);
	LargeObjectDesc *fc_loDesc;
	int			fc_written PG_USED_FOR_ASSERTS_ONLY;

	lo_cleanup_needed = true;
	fc_loDesc = inv_open(fc_loOid, INV_WRITE, CurrentMemoryContext);

	/* 权限检查 */
	if (!lo_compat_privileges &&
		pg_largeobject_aclcheck_snapshot(fc_loDesc->id,
										 GetUserId(),
										 ACL_UPDATE,
										 fc_loDesc->snapshot) != ACLCHECK_OK)
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("permission denied for large object %u",
						fc_loDesc->id)));

	inv_seek(fc_loDesc, fc_offset, SEEK_SET);
	fc_written = inv_write(fc_loDesc, VARDATA_ANY(fc_str), VARSIZE_ANY_EXHDR(fc_str));
	Assert(fc_written == VARSIZE_ANY_EXHDR(fc_str));
	inv_close(fc_loDesc);

	PG_RETURN_VOID();
}
